package com.atguigu.bigdata.gmall.realtime.app.dws;

import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.bean.KeywordBean;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.function.KWSplit;
import com.atguigu.bigdata.gmall.realtime.util.FlinkSinkUtil;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/10/14 08:28
 */
public class Dws_01_DwsTrafficSourceKeywordPageViewWindow extends BaseSQLApp {
    public static void main(String[] args) {
        new Dws_01_DwsTrafficSourceKeywordPageViewWindow().init(
            4001,
            2,
            "Dws_01_DwsTrafficSourceKeywordPageViewWindow"
        );
        
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        
        // 1. 建立一个动态表与 dwd 页面日志关联
        tEnv.executeSql("create table dwd_traffic_page(" +
                            "page map<String, String>, " +
                            "ts bigint, " +
                            "et as to_timestamp_ltz(ts, 3), " +
                            "watermark for et as et - interval '3' second " +  // 定义时间属性: 事件时间. 类型必须是 timestamp(3)
                            ")" + SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRAFFIC_PAGE, "Dws_01_DwsTrafficSourceKeywordPageViewWindow"));
        
        
        // 2. 过滤搜索记录, 读取搜索关键词
        Table kwTable = tEnv.sqlQuery("select " +
                                          " page['item'] kw, " +
                                          " et " +
                                          " from dwd_traffic_page " +
                                          " where (page['last_page_id']='search' or page['last_page_id']='home')" +
                                          " and page['item_type']='keyword' " +
                                          " and page['item'] is not null ");
        tEnv.createTemporaryView("kw_table", kwTable);
        // 3. 对关键词进行分词
        tEnv.createTemporaryFunction("kw_split", KWSplit.class);
        Table keywordTable = tEnv.sqlQuery("select " +
                                               "keyword, " +
                                               "et " +
                                               "from kw_table " +
                                               "join lateral table(kw_split(kw)) on true");
        /*tEnv.sqlQuery("select " +
                          "keyword, " +
                          "et " +
                          "from kw_table " +
                          ", lateral table(kw_split(kw))").execute().print();*/
        
        tEnv.createTemporaryView("keyword_table", keywordTable);
        // 4. 统计每次后的每个词的出现的次数: 分组开窗聚和
        /*tEnv.sqlQuery("select " +
                          " tumble_start(et, interval '5' second) stt, " +
                          " tumble_end(et, interval '5' second) edt, " +
                          " keyword, " +
                          " count(keyword) " +  // count(*) count(1) count(id) sum(1)
                          "from keyword_table " +
                          "group by keyword, tumble(et, interval '5' second)")
            .execute()
            .print();*/
        
        Table result = tEnv.sqlQuery("select " +
                                         " date_format(window_start, 'yyyy-MM-dd HH:mm:ss') stt, " +
                                         " date_format(window_end, 'yyyy-MM-dd HH:mm:ss') edt, " +
                                         " 'search' source, " +
                                         " keyword, " +
                                         " count(keyword) keyword_count, " +
                                         " unix_timestamp() * 1000 ts " +
                                         "from table( tumble( table keyword_table, descriptor(et), interval '5' second ) ) " +
                                         "group by keyword, window_start, window_end");
        // 5. 写出到 clickhouse 中
        // 把 table 转成流
        tEnv
            .toRetractStream(result, KeywordBean.class)
            .filter(t -> t.f0)
            .map(t -> t.f1)
            .addSink(FlinkSinkUtil.getClickHouseSink("dws_traffic_source_keyword_page_view_window", KeywordBean.class));
    
    
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    
    }
}
/*
分词:

"手机小米手机256g小米手机"

             手机
             小米
             手机
             256g
             小米
             手机
 
自定义函数: scalar(标量函数 一进一出)
          table(制表函数  一进多出)
          aggregate(聚合函数  多进一出)
          tableAggregate(制表聚合函数  多进多出)
          
原表  制出来的表 如何 在一起?
     靠 join

------------
统计用户搜索的关键词
热词榜, 搜索的热词

数据源:
    dwd_page 日志
    
什么样的数据:
    用户搜索记录
        last_page_id=search or last_page_id = home
        item_type = keyword
        item is not null

取出搜索关键词:
    item
    
"手机小米手机256g小米手机"
"手机华为手机"

 分词之后再统计集合   给中文分词的分词器: IK 分词器
 手机
 小米
 手机
 256g
 小米
 手机
 
 手机
 华为
 手机

-----
keyBy: 分词后的词
window: 基于事件事件的滚动窗口
    grouped window
        滚动
        select
            keyword,
            tumble_start(时间属性, 窗口长度)  stt
            tumble_end(时间属性, 窗口长度)  edt
            count(keyword)
        from t
        group by keyword, tumble(时间属性, 窗口长度)
        
        
        
        滑动
        session
        
    over window
        基于行
        基于时间


----
写出到 clickhouse 中:
    使用 clickhouse 连接器 没有
    
    自定义 sink.
    把表转成流之后,再自定义 sink 写入
    
    在 jdbc 的基础上, 封装一个 clickhouse sink

-
小批量汇总 轻度汇总 小窗口汇总

0-5   10
5-10  10
10-15 20
...

支持 sql 的数据库, 支持实时, 吞吐量
clickhouse

doris

tvf


*/